package com.navinfo.tripanalysis.service;

import com.navinfo.tripanalysis.pojo.Point0F37;
import com.navinfo.tripanalysis.pojo.LoadDataParam;
import com.navinfo.tripanalysis.pojo.Point;
import com.navinfo.tripanalysis.util.CommonUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

import java.util.*;

/**
 * 加载0F37协议数据的基本服务
 * @author 沈东生
 */
public abstract class AbstractLoad0F37DataServiceBigData implements Load0F37DataService {
    /**
     * 协议中是否含有protocoltype字段
     */
    protected  boolean supportProtocolType;
    private final String PROTOCOL_NAME = "protocoltype";

    protected static final List<String> COL_NAMES = new ArrayList<>();

    public AbstractLoad0F37DataServiceBigData() {
        COL_NAMES.addAll(Arrays.asList("gpstime", "latitude", "longitude", "height","gpsdirection", "engineoutputtorque", "speed", "accelerator", "brake", "rotation",  "clutchswitch", "realtimeoilconsumption", "fuelconsumptionrate","atmospherictemperature","instrumentspeed","ureainjection","urealevel"));
    }

    public boolean isSupportProtocolType() {
        return supportProtocolType;
    }

    public void setSupportProtocolType(boolean supportProtocolType) {
        if (supportProtocolType) {
            if (!COL_NAMES.contains(PROTOCOL_NAME)) {
                COL_NAMES.add(PROTOCOL_NAME);
            }
        } else {
            if (COL_NAMES.contains(PROTOCOL_NAME)) {
                COL_NAMES.remove(PROTOCOL_NAME);
            }
        }
        this.supportProtocolType = supportProtocolType;
    }

    /**
     * 获取过滤条件
     * @param param
     * @return
     */
    protected String getWhereSql0f37(LoadDataParam param) {
        String whereSql = "gpstime != 0 and gpstime is not null and tid != 0 and tid is not null ";

        List<Long> tids = param.getTids();
        if (CollectionUtils.isNotEmpty(tids)) {
            StringBuilder sb = new StringBuilder();
            sb.append(String.format(" and tid in(%s) ", CommonUtils.mkString(tids, ",")));
            whereSql += sb.toString();
        }

        //过滤出原0f37类型
        if(isSupportProtocolType()){
            whereSql += " and ( protocoltype is null or protocoltype=0 ) ";
        }

        return whereSql;
    }

    protected JavaRDD<Point> toJavaRdd(Dataset<Row> dataset) {
        return dataset.toJavaRDD().mapPartitions((FlatMapFunction<Iterator<Row>, Point>) iterator ->
                new Iterator<Point>() {
                    @Override
                    public boolean hasNext() {
                        return iterator.hasNext();
                    }
                    @Override
                    public Point next() {
                        return getPoint(iterator.next());
                    }
                }
        );
    }

    private Point getPoint(Row row) {
        Point0F37 point = new Point0F37();
        point.setTid(row.getAs("tid"));
        point.setProtocol(Point.PROTOCOL_0F37);
        point.setGpsTime(row.getAs("gpstime"));
        point.setLatitude(row.getAs("latitude"));
        point.setLongitude(row.getAs("longitude"));
        point.setDirection((int) row.getAs("gpsdirection")/100);
        point.setHeight(((int) row.getAs("height"))*1.0);
        point.setEngineOutputTorque(((int) row.getAs("engineoutputtorque")) / 100);
        point.setSpeedKmh(((int) row.getAs("speed")) / 100.0);
        point.setAccPedalPos(((int) row.getAs("accelerator"))/100);
        point.setRotation(((int) row.getAs("rotation")) / 100);
        point.setClutchSwitch(((int) row.getAs("clutchswitch")) / 100);
        point.setRealTimeOilConsumption(((int) row.getAs("realtimeoilconsumption")) / 100.0);
        point.setFuelConsumptionRate(((int) row.getAs("fuelconsumptionrate")) / 100.0);
        //0f37特有的属性
        point.setBrake(((int) row.getAs("brake")) / 100);
        point.setAtmosphericTem(((int) row.getAs("atmospherictemperature")) / 100);
        point.setInstrumentSpeed(((int) row.getAs("instrumentspeed")) / 100.0);
        point.setUreaInjection(((int) Optional.ofNullable(row.getAs("ureainjection")).orElse(0)));
        point.setUreaLevel(((int) Optional.ofNullable(row.getAs("urealevel")).orElse(0)));

        return point;
    }
}